package com.example.demo.service.impl;

import com.example.demo.dao.DzdpSkuMapper;
import com.example.demo.dao.SkuDao;
import com.example.demo.dao.SkuFailedDao;
import com.example.demo.dao.SkuFailedMapper;
import com.example.demo.domain.DzdpSku;
import com.example.demo.domain.DzdpSkuExample;
import com.example.demo.domain.SkuFailed;
import com.example.demo.domain.SkuFailedExample;
import com.example.demo.service.DZDPScratchService;
import com.example.demo.service.SkuService;
import com.example.demo.thread.RerunSkuTask;
import com.example.demo.thread.SkuTask;
import com.xiaoleilu.hutool.util.CollectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author ZhengWei
 * @Description
 * @date 2020/7/16 6:38 下午
 */
@Service
@Transactional(propagation = Propagation.REQUIRES_NEW)
public class SkuServiceImpl implements SkuService {
    @Value("${myThreadNum}")
    private int threadNum;
    @Autowired
    private DZDPScratchService scratchService;
    @Autowired
    private DzdpSkuMapper skuMapper;
    @Autowired
    private SkuFailedMapper skuFailedMapper;
    @Autowired
    private SkuDao skuDao;
    @Autowired
    private SkuFailedDao skuFailedDao;

    private Logger logger = LoggerFactory.getLogger(SkuServiceImpl.class);

    @Override
    public void getSkuList(Integer cityId) {
        //总任务数，同时作为爬虫线程共享信号量 在每个请求线程内共享
        AtomicInteger totalSku = new AtomicInteger(500000);

        long startTime = System.currentTimeMillis();

        //多线程爬取 sku 并存入内存
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 20, 1, TimeUnit.HOURS, new ArrayBlockingQueue<>(10));
        CountDownLatch countDownLatch = new CountDownLatch(threadNum);

        for (int threadCode = 1; threadCode <= threadNum; threadCode++) {
            SkuTask skuTask = new SkuTask(scratchService, cityId, totalSku,countDownLatch);
            threadPoolExecutor.submit(skuTask);
        }
//        while (totalSku.get() > 0) {}
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        for (int i = 0; i < 3; i++) {
            this.rerunFailed(cityId, threadPoolExecutor);
        }

        logger.info("-------------------------------------");
        logger.info("城市ID：" + cityId);
        logger.info("被禁 " + DZDPScratchImpl.forbiddenSkuIDSet.size() + " 条请求");

        this.writeNewSkuToDB();
        this.writeFailedToDB(cityId);

        //记录一下统计数据
        long end = System.currentTimeMillis();
        long s = (end - startTime) / 1000;
        long m = s / 60;
        long h = m / 60;
        logger.info("花费：" + h + " 小时 " + m % 60 + " 分钟 " + s % 60 + " 秒");
    }

    private void writeNewSkuToDB() {
        if (CollectionUtil.isNotEmpty(DZDPScratchImpl.skuMap)) {
            List<DzdpSku> values = new ArrayList<>(DZDPScratchImpl.skuMap.values());
            DzdpSkuExample skuExample = new DzdpSkuExample();
            skuExample.createCriteria().andIdIsNotNull();
            Set<Integer> skuIdExist = skuMapper.selectByExample(skuExample).stream().map(o -> o.getSkuid()).collect(Collectors.toSet());

            values = values.stream().filter(o -> !skuIdExist.contains(o.getSkuid())).collect(Collectors.toList());
            if (CollectionUtil.isNotEmpty(values)) {
                skuDao.batchInsert(values);
                logger.info("新增 " + values.size() + " 个菜品");
            }
            DZDPScratchImpl.skuMap.clear();
        }
    }

    private void writeFailedToDB(Integer cityId) {
        SkuFailedExample skuFailedExample = new SkuFailedExample();
        skuFailedExample.createCriteria().andIdIsNotNull();
        Set<Integer> skuFailedExistSet = skuFailedMapper.selectByExample(skuFailedExample).stream().map(o -> o.getSkuid()).collect(Collectors.toSet());
        List<Integer> skuFailed = DZDPScratchImpl.forbiddenSkuIDSet.stream().filter(o -> !skuFailedExistSet.contains(o)).collect(Collectors.toList());
        if (CollectionUtil.isNotEmpty(skuFailed)) {
            skuFailedDao.batchInsert(skuFailed, cityId);
            DZDPScratchImpl.forbiddenSkuIDSet.clear();
        }
    }

    private void rerunFailed(Integer cityId, ThreadPoolExecutor threadPoolExecutor) {
        ArrayList<Integer> skuIdList = new ArrayList<>(DZDPScratchImpl.forbiddenSkuIDSet);
        DZDPScratchImpl.forbiddenSkuIDSet.clear();
        AtomicInteger index = new AtomicInteger(skuIdList.size());
        for (int i = 1; i <= threadNum; i++) {
            RerunSkuTask rerunSkuTask = new RerunSkuTask(scratchService, cityId, skuIdList, index);
            threadPoolExecutor.submit(rerunSkuTask);
        }
        while (index.get() > -1) {}
    }
}
